package com.muchfish.reactor.api;

import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;

import java.time.Duration;

public class ReactorConnectableFluxExample {
    public static void main(String[] args) throws InterruptedException {
        ConnectableFlux<Integer> connectableFlux = Flux.range(1, 3)
                .delayElements(Duration.ofSeconds(1))
                .publish(); // 将普通的 Flux 转换为 ConnectableFlux

        connectableFlux.connect(); // 开始生成元素

        // 第一个订阅者
        connectableFlux.subscribe(data -> System.out.println("Subscriber 1: " + data));

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 第二个订阅者，共享相同的序列
        connectableFlux.subscribe(data -> System.out.println("Subscriber 2: " + data));

        Thread.sleep(23333);
        // 结果：
        // Subscriber 1: 1
        // Subscriber 1: 2
        // Subscriber 2: 2
        // Subscriber 1: 3
        // Subscriber 2: 3
    }
}
